package com.bangcommunity.bbframe.mq.kafka.consumer;

import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.io.PropertiesUtil;
import com.bangcommunity.bbframe.common.utils.lang.SpiUtils;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.dynconfig.DynConfigRegister;
import com.bangcommunity.bbframe.dynconfig.IDynConfigProcessor;
import com.bangcommunity.bbframe.mq.IQtMessager;
import com.bangcommunity.bbframe.mq.consumer.IQtMsgProcessor;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class KafkaConsumerClient {
    static Logger logger = LoggerFactory.getLogger(KafkaConsumerClient.class);

    private KafkaConsumerClient() {
    }

    private static class KafkaProducerClientHolder {
        private static DelegateKafkaConsumer instance;
        private static Properties properties;
        private static final String producer_config_group = "common";
        private static final String producer_config_dataId = "kafkaConsumer";
        private static final String productCode = MainConfigLoader.getInstance().getProductCode();
        private static final String appCode = MainConfigLoader.getInstance().getAppCode();
        static {
            DynConfigRegister.getInstance().buildOnce(productCode, appCode, producer_config_group,
                    producer_config_dataId, new IDynConfigProcessor() {
                        @Override
                        public void procesConfig(String config) {
                            properties = PropertiesUtil.loadFromString(config);
                        }
                    });
        }

        /**
         * 属性基于0.10.1
         */
        private static DelegateKafkaConsumer buildKafkaConsumer(String rawTopic, IQtMsgProcessor msgProcessor,
                Properties prop) {
            if (null == prop) {
                prop = new Properties();
            }
            Properties allProperties = new Properties();
            allProperties.putAll(properties);
            allProperties.putAll(prop);
            // host/port对的列表，用来建立与kafka的初始链接。
            // setDefault(allProperties, "bootstrap.servers", "");//default:
            // consumer offsets可以自动提交到kafka的频率（微秒）
            // setDefault(allProperties,"auto.commit.interval.ms","");//default:5000
            // 当kafka没有初始offset或者server中也不存在任何初始化offset时，consumer遇到这种情况应该从哪里开始获取消息
            // setDefault(allProperties,"auto.offset.reset","");//default:latest
            // 自动检查消费的消息的CRC32.这个检查保证了消息在发送过程中没有损坏，或者在磁盘上没有损坏。
            // 这个检查有可能增加负担，因此对性能要求比较高的情况可能禁用这个检查。
            // setDefault(allProperties,"check.crcs","");//default:true
            //
            setDefault(allProperties, "client.id", productCode + "_" + appCode);// default:
            // 空闲链接的超时时间：server socket处理线程会关闭超时的链接。
            // setDefault(allProperties,"connections.max.idle.ms","");//default:540000
            // 如果设置为true，则consumer的offset会在后台周期性的上传
            // setDefault(allProperties,"enable.auto.commit","");//default:true
            // 内部topics（例如offsets）是否需要暴漏给consumer。如果设置true，从内部topic获取数据的方式只能是订阅它。
            // setDefault(allProperties,"exclude.internal.topics","");//default:true
            // server针对抓取请求的应答中所包含的最大字节数。
            // setDefault(allProperties,"fetch.max.bytes","");//default:52428800
            // server在应答抓取请求之前可以阻塞的最长时间
            // setDefault(allProperties,"fetch.max.wait.ms","");//default:500
            // server返回给抓取请求的最小数据量。
            // setDefault(allProperties,"fetch.min.bytes","");//default:1
            // 代表consumer组的唯一字符串
            setDefault(allProperties, "group.id", StringUtils.join(new Object[] { productCode, appCode }, "_"));// default:""
            // 当使用Kafka的group管理用法时，consumer协作器两次心跳之间的时间间隔。
            // setDefault(allProperties,"heartbeat.interval.ms","");//default:3000
            // server返回消息中针对每个partition数据请求的最大数据量。
            // setDefault(allProperties,"max.partition.fetch.bytes","");//default:1048576
            // 当使用consumer 组管理时，在两次调用poll（）之间的停留时间。
            // setDefault(allProperties,"max.poll.interval.ms","");//default:300000
            // 一次单独调用poll（）可以返回的消息的最大条数。
            // setDefault(allProperties,"max.poll.records","");//default:500
            // 更新metadata的时间间隔，
            // setDefault(allProperties,"metadata.max.age.ms","");//default:300000
            // 用于实现指标统计的类的列表
            // setDefault(allProperties,"metric.reporters","");//default:
            // 维护计算指标的样本数
            // setDefault(allProperties,"metrics.num.samples","");//default:2
            // 度量样本的计算的时长
            // setDefault(allProperties,"metrics.sample.window.ms","");//default:30000
            // partitions的分配策略的类名。当使用group管理策略时，客户端用来将来将partitions分配给组中consumer实例。
            // setDefault(allProperties,"partition.assignment.strategy","");//default:org.apache.kafka.clients.consumer.RangeAssignor
            // TCP接受缓存的大小（SO_RCVBUF）。如果设置为－1，则使用OS默认值.
            // setDefault(allProperties,"receive.buffer.bytes","");//default:65536
            // 重连给定host之前的等待时间。避免频繁的重连某个host。
            // setDefault(allProperties,"reconnect.backoff.ms","");//default:50
            // 客户端等待broker应答的超时时间。
            // setDefault(allProperties,"request.timeout.ms","");//default:305000
            // 重新发送失败请求给某个topic partition之前的最长等待时间，避免极短时间内频繁的重试。
            // setDefault(allProperties,"retry.backoff.ms","");//default:100
            // TCP发送的socket的SO_SNDBUF缓存。如果设置为－1，将使用OS的默认值
            // setDefault(allProperties,"send.buffer.bytes","");//default:131072
            // 当使用Kafka group管理用法时，这个超时时间用来检测consumer是否失效
            // setDefault(allProperties,"session.timeout.ms","");//default:10000
            /**
             * 以下参数不允许配置
             */
            // Serializer接口的密钥的类的key
            overrideValue(allProperties, "key.deserializer",
                    "com.bangcommunity.bbframe.mq.kafka.serial.QtMessageDeserializer");// defalult:
            // Serializer接口的类的value
            overrideValue(allProperties, "value.deserializer",
                    "com.bangcommunity.bbframe.mq.kafka.serial.QtMessageDeserializer");// defalult:
            List<ConsumerInterceptor> consumerInterceptors = SpiUtils.listSpi(ConsumerInterceptor.class);
            StringBuffer intecptStr = new StringBuffer();
            if (null != consumerInterceptors) {
                for (int i = 0; i < consumerInterceptors.size(); i++) {
                    if (i != 0) {
                        intecptStr.append(",");
                    }
                    intecptStr.append(consumerInterceptors.get(i).getClass().getName());
                }
            }
            // 用作拦截器的类的列表。
            overrideValue(allProperties, "interceptor.classes", intecptStr.toString());// defalult:
            return new DelegateKafkaConsumer(rawTopic, new KafkaConsumer<String, IQtMessager>(allProperties),
                    msgProcessor);
        }

        private static void setDefault(Properties properties, String key, Object defalt) {
            if (StringUtils.isNotBlank(key) && null != properties && null != defalt) {
                Object o = properties.get(key);
                if (null == o) {
                    o = defalt;
                }
                properties.put(key, o);
            }
        }

        private static void overrideValue(Properties properties, String key, Object newValue) {
            if (StringUtils.isNotBlank(key) && null != properties && null != newValue) {
                properties.put(key, newValue);
            }
        }
    }

    public static Runnable createConsumerTask(String rawTopic, IQtMsgProcessor msgProcessor, Properties prop) {
        DelegateKafkaConsumer delegateKafkaConsumer = KafkaProducerClientHolder.buildKafkaConsumer(rawTopic,
                msgProcessor, prop);
        if (null == delegateKafkaConsumer) {
            throw new RuntimeException("build kafka producer client exception");
        }
        return delegateKafkaConsumer;
    }

}
